package upload_task_new

import (
	"bytes"
	"context"
	"crypto/sha256"
	"crypto/tls"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net"
	"net/http"
	"os"
	"sync/atomic"
	"time"

	"gitee.com/baidugoogle/upload_file/upload_task/structs"
)

const PartSize = 16 * 1024 * 1024

type GetToken func() string

type Task struct {
	uri string
	f   GetToken
}

func NewTask(uri string, f GetToken) *Task {
	return &Task{
		uri: uri,
		f:   f,
	}
}

func (t *Task) Upload(filename string, info *structs.JsonCreateExt, threadCount int) (string, error) {

	cli := t.newHttpClient()

	var request = structs.JsonCreateRequest{
		PartSize: PartSize,
		Extra:    info,
	}

	fp, err := os.Open(filename)
	if err != nil {
		log.Printf("open file fail. %v [%v]", err, filename)
		return "", err
	} else {
		defer fp.Close()
	}

	// 准备
	{

		// 分页大小
		if fileStat, err := fp.Stat(); err != nil {
			return "", err
		} else {
			request.Name = fileStat.Name()
			request.PartCount = structs.PageCount(int(fileStat.Size()), PartSize)
			request.Size = int(fileStat.Size())
		}

		// 预哈希
		if true {
			buffer := make([]byte, 1024)
			_, _ = io.LimitReader(fp, 1024).Read(buffer)
			request.PreHash = structs.SHA256(buffer)

			_, _ = fp.Seek(0, io.SeekStart)
		}

		// 调试, 上传添加整个文件的Hash
		if true {
			if h, err := structs.SHA256File(filename); err != nil {
				log.Printf("SHA256File fail. %v [%v]", err, filename)
				return "", err
			} else {
				request.ContentHash = hex.EncodeToString(h)
			}
		}
	}

	// 创建上传任务
	objectCreateReply, err := t.httpCreate(cli, t.getURL("/fileUpload/create"), &request)
	if err != nil {
		log.Printf("httpCreate fail. %v [%v]", err, filename)
		return "", err
	}

	switch objectCreateReply.FileState {
	case structs.FileStateFileIdMatched, structs.FileStateContentMatched:
		// 文件匹配, 文件已存在, 秒传
		// 已完成, 结束
		log.Printf("秒传 %v [%v]", objectCreateReply.FileState, filename)
		return "", nil
	case structs.FileStatePreHashMatched:
		// 预哈希匹配, 下一步匹配文件哈希
		// 测试完整Hash
		//   1. 相同, 结束
		//   2. 不相同, 上传
		if h, err := structs.SHA256File(filename); err != nil {
			log.Printf("SHA256File fail. %v [%v]", err, filename)
			return "", err
		} else {
			request.ContentHash = hex.EncodeToString(h)
		}

		// 获取文件哈希, 判断是否可以秒传
		if objectCreateReply, err = t.httpCreate(cli, t.getURL("/fileUpload/create"), &request); err != nil {
			return "", err
		} else {
			switch objectCreateReply.FileState {
			case structs.FileStateContentMatched:
				// 文件内容匹配, 文件已存在, 秒传
				// 已完成, 结束
				log.Printf("秒传 [%v]", filename)
				return "", nil
			case structs.FileStateNewFile:
				// 新文件case
				// 上传
			default:
				return "", fmt.Errorf("unknown file state %v", objectCreateReply.FileState)
			}
		}

	case structs.FileStateNewFile:
		// 新文件case
		// 上传
	default:
		return "", fmt.Errorf("unknown file state %v", objectCreateReply.FileState)
	}

	if objectCreateReply.NewFile == nil {
		return "", fmt.Errorf("error. new file is nil")
	}

	uploadId := objectCreateReply.NewFile.UploadId

	log.Printf("upload part count %v [%v]", request.PartCount, filename)
	now := time.Now()

	chanParts := make(chan *ObjectPartInfo, 4)
	go func() {
		defer close(chanParts)

		h := sha256.New()
		for i := 0; i < request.PartCount; i++ {
			partNumber := i

			r := io.NewSectionReader(fp, int64(partNumber*PartSize), PartSize)

			h.Reset()
			rr := NewHookReader(r, func(r io.Reader, p []byte) (n int, err error) {
				n, err = r.Read(p)
				if n > 0 {
					h.Write(p[:n])
				}
				return
			})
			data, _ := io.ReadAll(rr)

			object := &ObjectPartInfo{
				cli:        cli,
				uploadId:   uploadId,
				partNumber: partNumber,
				data:       data,
				filename:   filename,
				partHash:   hex.EncodeToString(h.Sum(nil)),
			}

			// log.Printf("投递分片 [%v][%v]", filename, i)
			chanParts <- object
		}
	}()

	var replyPartCount atomic.Int32
	ctx, ctxFunc := context.WithCancel(context.TODO())
	defer ctxFunc()

	for i := 0; i < threadCount; i++ {
		index := i
		go func(index int) {
			cli := t.newHttpClient()

			for {
				object, ok := <-chanParts
				if !ok {
					return
				}

				object.cli = cli

				//log.Printf("上传分片: [%v]", object.partNumber)
				partTime := time.Now()
				reply, err := t.httpPut(object)
				if err != nil {
					log.Printf("上传分片结束 [%v][%02v/%v] 耗时:[%v] error: %v",
						object.filename, object.partNumber, request.PartCount, time.Since(partTime), err)
				} else {
					log.Printf("上传分片结束 [%v][%02v/%v] 耗时:[%v] msg: %v",
						object.filename, object.partNumber, request.PartCount, time.Since(partTime), reply.Msg)
				}

				replyPartCount.Add(1)
				if replyPartCount.Load() >= int32(request.PartCount) {
					ctxFunc()
				}
			}
		}(index)
	}
	select {
	case <-ctx.Done():
		break
	}
	log.Printf("耗时: %v [%v]", time.Since(now), filename)

	reply, err := t.httpComplete(cli, uploadId, filename)
	if err != nil {
		log.Printf("上传完成 [%v]: error: %v", filename, err)
		return "", err
	} else {
		log.Printf("上传完成 [%v]: msg: %v", filename, reply.Msg)
	}

	if reply.Code != structs.BVResultSuccess {
		return uploadId, fmt.Errorf("%v", reply.Msg)
	}

	return uploadId, nil
}

// 创建上传任务
func (t *Task) httpCreate(cli *http.Client, uri string, request *structs.JsonCreateRequest) (*structs.JsonCreateReply, error) {
	log.Println("[http create]")
	body, err := json.Marshal(request)
	if err != nil {
		return nil, err
	}

	reply, duration, err := structs.HttpRequest(cli, http.MethodPost, uri, bytes.NewBuffer(body), "application/json; charset=utf-8", t.getHeaders())
	if err != nil {
		return nil, err
	} else {
		log.Println(structs.FormatJsonString(reply))
	}

	var objectCreateReply structs.JsonCreateReply
	if err = json.Unmarshal(reply, &objectCreateReply); err != nil {
		log.Println(err)
		return nil, err
	} else if objectCreateReply.Code != structs.BVResultSuccess {
		return &objectCreateReply, fmt.Errorf("create fail. %v %v", objectCreateReply.Code, objectCreateReply.Msg)
	}

	objectCreateReply.Duration = duration
	return &objectCreateReply, nil
}

type ObjectPartInfo struct {
	cli        *http.Client
	uploadId   string
	partNumber int
	data       []byte
	filename   string
	partHash   string
}

// 上传分片
func (t *Task) httpPut(info *ObjectPartInfo) (*structs.JsonPutPartReply, error) {
	var (
		uri = fmt.Sprintf(t.getURL("/fileUpload/put/%v/%v"), info.uploadId, info.partNumber)

		partFilename = fmt.Sprintf("%v", info.partNumber)
		reply        []byte
		err          error
	)
	_ = partFilename

	reply, err = structs.NewMultipartFormData(info.cli, uri, nil, "file", partFilename, info.data, t.getHeaders(map[string]string{
		"Content-Hash": info.partHash,
	}))
	if err != nil {
		log.Printf("NewMultipartFormData fail. %v [%v]", err, info.filename)
		return nil, err
	} else {
		//log.Println(structs.FormatJsonString(reply))
	}

	var objectPutReply structs.JsonPutPartReply
	if err = json.Unmarshal(reply, &objectPutReply); err != nil {
		//objectPutReply.Duration = duration
		log.Printf("Unmarshal fail. %v [%v]", err, info.filename)
		return nil, err
	} else if objectPutReply.Code != structs.BVResultSuccess {
		//objectPutReply.Duration = duration
		return &objectPutReply, fmt.Errorf("%v", objectPutReply.Msg)
	}

	//objectPutReply.Duration = duration

	return &objectPutReply, nil
}

// 完成
func (t *Task) httpComplete(cli *http.Client, uploadId string, filename string) (*structs.JsonCompleteReply, error) {
	log.Printf("http complete [%v]", filename)

	var objectReply structs.JsonCompleteReply

	uri := fmt.Sprintf(t.getURL("/fileUpload/complete/%v"), uploadId)
	reply, duration, err := structs.HttpRequest(cli, http.MethodPost, uri, nil, "application/json; charset=utf-8", t.getHeaders())
	if err != nil {
		log.Printf("HttpRequest fail. %v [%v]", err, filename)
		objectReply.Duration = duration
		return &objectReply, err
	} else {
		log.Printf("HttpRequest result. [%v] %v", filename, structs.FormatJsonString(reply))
	}

	if err = json.Unmarshal(reply, &objectReply); err != nil {
		log.Printf("Unmarshal fail. %v [%v]", err, filename)
		objectReply.Duration = duration
		return &objectReply, err
	} else if objectReply.Code != structs.BVResultSuccess {
		objectReply.Duration = duration
		return &objectReply, fmt.Errorf("%v", objectReply.Msg)
	}

	objectReply.Duration = duration
	return &objectReply, nil
}

// HttpCancel 取消
func (t *Task) HttpCancel(uploadId string) (*structs.JsonCancelReply, error) {
	log.Println("[http cancel]")
	cli := t.newHttpClient()

	uri := fmt.Sprintf(t.getURL("/fileUpload/cancel/%v"), uploadId)
	reply, duration, err := structs.HttpRequest(cli, http.MethodPost, uri, nil, "application/json; charset=utf-8", t.getHeaders())
	if err != nil {
		log.Println(err)
		return nil, err
	} else {
		log.Println(structs.FormatJsonString(reply))
	}

	var objectReply structs.JsonCancelReply
	if err = json.Unmarshal(reply, &objectReply); err != nil {
		log.Println(err)
		objectReply.Duration = duration
		return &objectReply, err
	} else if objectReply.Code != structs.BVResultSuccess {
		objectReply.Duration = duration
		return &objectReply, fmt.Errorf("%v", objectReply.Msg)
	}

	objectReply.Duration = duration
	return &objectReply, nil
}

func (t *Task) getURL(path string) string {
	return t.uri + "/bvnru/v1" + path
}

func (t *Task) getHeaders(kvs ...map[string]string) map[string]string {
	h := map[string]string{
		"Authorization": t.f(),
	}

	for _, kv := range kvs {
		for k, v := range kv {
			h[k] = v
		}
	}

	return h
}

func (t *Task) newHttpClient() *http.Client {
	dialer := &net.Dialer{
		Timeout:   30 * time.Second,
		KeepAlive: 30 * time.Second,
	}

	cli := &http.Client{
		Transport: &http.Transport{
			TLSClientConfig: &tls.Config{
				InsecureSkipVerify: true,
			},
			DialContext:           dialer.DialContext,
			MaxIdleConns:          256,
			MaxIdleConnsPerHost:   16,
			ResponseHeaderTimeout: time.Minute,
			IdleConnTimeout:       time.Minute,
			TLSHandshakeTimeout:   10 * time.Second,
			ExpectContinueTimeout: 10 * time.Second,
			// Set this value so that the underlying transport round-tripper
			// doesn't try to auto decode the body of objects with
			// content-encoding set to `gzip`.
			//
			// Refer:
			//    https://golang.org/src/net/http/transport.go?h=roundTrip#L1843
			DisableCompression: true,
		},
	}

	return cli
}

type ReaderFunc func(p []byte) (n int, err error)

func (f ReaderFunc) Read(p []byte) (n int, err error) {
	return f(p)
}

// NewHookReader hook reader
//
//	h := sha256.New()
//	NewHookReader(nil, func(r io.Reader, p []byte) (n int, err error) {
//		if n, err = r.Read(p); n > 0 {
//			h.Write(p[:n])
//		}
//		return
//	})
//	_ = hex.EncodeToString(h.Sum(nil))
func NewHookReader(r io.Reader, hook func(r io.Reader, p []byte) (n int, err error)) io.Reader {
	type object struct {
		io.Reader
		hook func(r io.Reader, p []byte) (int, error)
	}

	return &object{
		hook: hook,
		Reader: ReaderFunc(func(p []byte) (n int, err error) {
			return hook(r, p)
		}),
	}
}
